package redis.inaction.clients.ops;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.reactive.RedisReactiveCommands;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import io.lettuce.core.cluster.api.reactive.RedisAdvancedClusterReactiveCommands;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@SuppressWarnings("ALL")
@Slf4j
public class LettuceOps extends AbstractOps{

    public static final String SYNC_MODE = "sync";
    public static final String ASYNC_MODE = "async";
    public static final String REACTIVE_MODE = "reactive";
    /**
     * 一般不建议配置 required = false, 此处是演示需要
     */
    @Autowired(required = false)
    private RedisClusterClient redisClusterClient;

    @Autowired(required = false)
    private RedisClient redisClient;

    @Value("${lettuce.mode:sync}")
    private String mode;

    private StatefulRedisClusterConnection<String, String> sharedClusterConnection;
    private StatefulRedisConnection<String, String> sharedConnection;

    public boolean isLettuceInitialized(){
        return sharedClusterConnection != null || sharedConnection != null;
    }

    public void doSomething(){
        if(sharedConnection != null){
            standaloneOps();
        }

        if(sharedClusterConnection != null ){
            clusterOps();
        }
    }

    protected void standaloneOps(){
        log.info("lettuce standalone ops:");
        if( SYNC_MODE.equalsIgnoreCase(mode) ){
            syncOps();
        } else if (ASYNC_MODE.equalsIgnoreCase(mode)){
            asyncOps();
        } else if (REACTIVE_MODE.equalsIgnoreCase(mode)){
            reactiveOps();
        } else {
            log.error("Please add config: lettuce.mode");
        }
    }

    private void syncOps(){
        log.info("mode: sync");
        String hashKey = "lettuce_sync_demo";
        RedisCommands<String,String> command = sharedConnection.sync();
        //clean
        List<String> fields = command.hkeys(hashKey);
        if(!CollectionUtils.isEmpty(fields)){
            Long delKeys = command.hdel(hashKey, fields.toArray(new String[0]));
            log.info("hdel {} {} fields: {}", hashKey, delKeys, String.join(",", fields));
        }
        //hset, hgetall, hget
        redisOps(
                index -> {
                    boolean rt = command.hset(hashKey, "k_" + index, "value_" + index);
                    log.info("hset {}-{}={}: {}", hashKey, ("k_" + index), ("value_" + index),  rt);
                },
                command::hgetall,
                command::hget,
                hashKey);
    }

    private void asyncOps(){
        log.info("mode: async");
        String hashKey = "lettuce_ASYNC_demo";
        RedisAsyncCommands<String,String> asyncCommand = sharedConnection.async();
        //clean
        RedisFuture<List<String>> fieldsFuture = asyncCommand.hkeys(hashKey);
        List<String> fields = getFutureResult(fieldsFuture);
        String[] fieldsArr = fields == null ? new String[0] : fields.toArray(new String[0]);
        asyncCommand.hdel(hashKey, fieldsArr).thenAccept(
                delKeys -> log.info("hdel {} {} fields: {}", hashKey, delKeys, String.join(",", fieldsArr))
        );
        //hset, hgetall, hget
        redisOps(index -> asyncCommand.hset(hashKey, "k_" + index, "value_" + index)
                        .thenAccept(rt -> log.info("hset {}-{}={}: {}", hashKey, ("k_" + index), ("value_" + index),  rt)),
                 key -> {
                     RedisFuture<Map<String,String>> f = asyncCommand.hgetall(key);
                     return getFutureResult(f);
                 },
                (key, field) -> {
                    RedisFuture<String> f = asyncCommand.hget(key,field);
                    return getFutureResult(f);
                },
                hashKey);
    }

    private void reactiveOps(){
        log.info("mode: reactive");
        String hashKey = "lettuce_reactive_demo";
        RedisReactiveCommands<String,String> reactiveCommand = sharedConnection.reactive();
        //clean
        reactiveCommand.hkeys(hashKey).buffer().subscribe(fields -> {
            reactiveCommand.hdel(hashKey, fields.toArray(new String[0]))
                    .subscribe(delKeys -> {
                        log.info("hdel {} {} fields: {}", hashKey, delKeys, String.join(",", fields.toArray(new String[0])));
                    });
        });

        //hset, hgetall, hget
        redisOps(
                index -> reactiveCommand.hset(hashKey, "k_" + index, "value_" + index)
                        .doOnSuccess( rt -> log.info("hset {}-{}={}: {}", hashKey, ("k_" + index), ("value_" + index),  rt))
                        .subscribe(),
                key -> {
                    Map<String,String> map = new HashMap<>();
                    reactiveCommand.hgetall(key)
                            .doOnNext( kv -> map.put(kv.getKey(), kv.getValue()))
                            .doOnComplete(() -> log.info("Completed"))
                            .subscribe();
                    return map;
                },
                (key,field) -> reactiveCommand.hget(key,field).block(),
                hashKey
        );
    }

    protected void clusterOps(){
        log.info("lettuce Cluster ops:");
        if( SYNC_MODE.equalsIgnoreCase(mode) ){
            syncClusterOps();
        } else if (ASYNC_MODE.equalsIgnoreCase(mode)){
            asyncClusterOps();
        } else if (REACTIVE_MODE.equalsIgnoreCase(mode)){
            reactiveClusterOps();
        } else {
            log.error("Please add config: lettuce.mode");
        }
    }

    private void syncClusterOps(){
        log.info("mode: sync");
        String hashKey = "lettuce_cluster_sync_demo";
        RedisAdvancedClusterCommands<String,String> clusterCommand = sharedClusterConnection.sync();
        //clean
        List<String> fields = clusterCommand.hkeys(hashKey);
        if(!CollectionUtils.isEmpty(fields)){
            Long delKeys = clusterCommand.hdel(hashKey, fields.toArray(new String[0]));
            log.info("hdel {} {} fields: {}", hashKey, delKeys, String.join(",", fields));
        }
        //hset, hgetall, hget
        redisOps(
                index -> {
                    boolean rt = clusterCommand.hset(hashKey, "k_" + index, "value_" + index);
                    log.info("hset {}-{}={}: {}", hashKey, ("k_" + index), ("value_" + index),  rt);
                },
                clusterCommand::hgetall,
                clusterCommand::hget,
                hashKey);
    }

    private void asyncClusterOps(){
        log.info("mode: async");
        String hashKey = "lettuce_cluster_ASYNC_demo";
        RedisAdvancedClusterAsyncCommands<String,String> asyncClusterCommand = sharedClusterConnection.async();
        //clean
        RedisFuture<List<String>> fieldsFuture = asyncClusterCommand.hkeys(hashKey);
        List<String> fields = getFutureResult(fieldsFuture);
        String[] fieldsArr = fields == null ? new String[0] : fields.toArray(new String[0]);
        asyncClusterCommand.hdel(hashKey, fieldsArr).thenAccept(
                delKeys -> log.info("hdel {} {} fields: {}", hashKey, delKeys, String.join(",", fieldsArr))
        );

        //hset, hgetall, hget
        redisOps(index -> asyncClusterCommand.hset(hashKey, "k_" + index, "value_" + index)
                        .thenAccept(rt -> log.info("hset {}-{}={}: {}", hashKey, ("k_" + index), ("value_" + index),  rt)),
                key -> {
                    RedisFuture<Map<String,String>> f = asyncClusterCommand.hgetall(key);
                    return getFutureResult(f);
                },
                (key, field) -> {
                    RedisFuture<String> f = asyncClusterCommand.hget(key,field);
                    return getFutureResult(f);
                },
                hashKey);
    }

    private void reactiveClusterOps(){
        log.info("mode: reactive");
        String hashKey = "lettuce_cluster_reactive_demo";
        RedisAdvancedClusterReactiveCommands<String,String> reactiveClusterCommand = sharedClusterConnection.reactive();

        //clean
        reactiveClusterCommand.hkeys(hashKey).buffer().subscribe(fields -> {
            reactiveClusterCommand.hdel(hashKey, fields.toArray(new String[0]))
                    .subscribe(delKeys -> {
                        log.info("hdel {} {} fields: {}", hashKey, delKeys, String.join(",", fields.toArray(new String[0])));
                    });
        });

        //hset, hgetall, hget
        redisOps(
                index -> reactiveClusterCommand.hset(hashKey, "k_" + index, "value_" + index)
                        .doOnSuccess( rt -> log.info("hset {}-{}={}: {}", hashKey, ("k_" + index), ("value_" + index),  rt))
                        .subscribe(),
                key -> {
                    Map<String,String> map = new HashMap<>();
                    reactiveClusterCommand.hgetall(key)
                            .doOnNext(kv -> map.put(kv.getKey(),kv.getValue()))
                            .doOnComplete(() -> log.info("Completed"))
                            .subscribe();
                    return map;
                },
                (key,field) -> reactiveClusterCommand.hget(key,field).block(),
                hashKey
        );
    }

    private <T> T getFutureResult(RedisFuture<T> f){
        try {
            return f == null ? null : f.get(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("return null since exception occurred.", e);
            return null;
        }
    }

    private void onSuccess(Object t){
    }

    @PostConstruct
    public void init(){
        if(redisClusterClient != null){
            sharedClusterConnection = redisClusterClient.connect();
        }
        if( redisClient != null ){
            sharedConnection = redisClient.connect();
        }
    }

    @PreDestroy
    public void close(){
        if(sharedConnection != null){
            sharedConnection.close();
        }

        if(sharedClusterConnection != null){
            sharedClusterConnection.close();
        }
    }
}
